（以 FastAPI + LINE Messaging API v3 + yfinance + Cloudinary + gTTS/OpenAI TTS + Groq 為例）
https://ithelp.ithome.com.tw/ironman

今天把兩個看似小、但實際最影響體感的功能補齊：輸入幣別時要走「查匯」而非美股代碼、每一則回覆都要附 TTS 語音。這兩件事會同時牽涉「意圖判斷 → 路由」與「訊息編排 → 多訊息回覆」，也最容易踩 API 互動的坑（例如把同步 API 誤用成 await）。我會從規格定義、程式設計、除錯與驗收一路講到位。
https://developers.line.biz/en/docs/messaging-api/overview/

規格：輸入一段文字 → 決定要走「外匯」或「股票」
• 若輸入是 3 碼幣別（如 JPY）或自然語詞（如「日圓」），優先走外匯（Yahoo Finance 外匯代碼規則為 BASEQUOTE=X，例如 USDJPY=X）。
• 若輸入為台股 4–6 位數字（可帶尾碼）或 1–5 位英文字母（美股代碼），走股票分析（StockGPT）。
• 特別處理：僅單一幣別時，預設以 FX_DEFAULT_QUOTE=TWD（可用環境變數調整）。
https://finance.yahoo.com/ ・ https://github.com/ranaroussi/yfinance ・ https://docs.python.org/3/library/re.html

規格：每次回覆都附「主選單 Quick Reply」與「語音（TTS）」
• Quick Reply 第一顆固定「主選單」，再附常用指令（台股/美股/金價/JPY/2330/NVDA…）。
• TTS 支援 TTS_PROVIDER=auto|openai|gtts，auto 會先試 OpenAI TTS，不行就 fallback gTTS；音檔上傳 Cloudinary，回傳 AudioMessage。
• 可以用 TTS_SEND_ALWAYS=true/false 控制是否每則都附語音。
https://developers.line.biz/en/docs/messaging-api/using-quick-reply/ ・ https://platform.openai.com/docs/guides/text-to-speech ・ https://pypi.org/project/gTTS/ ・ https://cloudinary.com/documentation/image_and_video_upload_api_reference

路由判斷：外匯優先的 Regex 與 Alias

以下整理我在專案中的實作（完整、不省略，新增或修改處均以 # [NEW] / # [CHANGED] 註解）——這一段專責把「查匯」從股票分流出來：
https://docs.python.org/3/library/re.html

[NEW] 幣別清單與別名

FX_CODES = {"USD","TWD","JPY","EUR","GBP","CNY","HKD","AUD","CAD","CHF","SGD","KRW","NZD","THB","MYR","IDR","PHP","INR","ZAR"}
FX_ALIAS = {"日圓":"JPY","日元":"JPY","美元":"USD","台幣":"TWD","新台幣":"TWD","人民幣":"CNY","港幣":"HKD","韓元":"KRW","歐元":"EUR","英鎊":"GBP"}
FX_DEFAULT_QUOTE = os.getenv("FX_DEFAULT_QUOTE", "TWD").upper()

def is_fx_query(text: str) -> bool:
t = text.strip().upper()
if t in FX_CODES or t in FX_ALIAS.values():
return True
# 允許 "USD/JPY", "USD JPY", "USDJPY"
return bool(re.match(r"^[A-Z]{3}[\s/-]?([A-Z]{3})?$", t))

def _normalize_fx_token(tok: str) -> str:
tok = tok.strip().upper()
return FX_ALIAS.get(tok, tok)

def parse_fx_pair(user_text: str) -> tuple[str, str, str]:
# 輸入可能是「JPY」或「USD/JPY」或「美元 日圓」
raw = user_text.strip()
m = re.findall(r"[A-Za-z\u4e00-\u9fa5]{2,5}", raw)
toks = [_normalize_fx_token(x) for x in m]
toks = [x for x in toks if x in FX_CODES]
if not toks:
# 單一 token，但正好是 3 碼幣別
t = _normalize_fx_token(raw)
if len(t) == 3 and t in FX_CODES:
base, quote = t, FX_DEFAULT_QUOTE
else:
base, quote = "USD", "JPY" # fallback
elif len(toks) == 1:
base, quote = toks[0], FX_DEFAULT_QUOTE
else:
base, quote = toks[0], toks[1]
symbol = f"{base}{quote}=X"
link = f"https://finance.yahoo.com/quote/{symbol}/"
return base, quote, link

上面 parse_fx_pair() 最後組的是 Yahoo Finance 的外匯代碼（=X 後綴），這能直接餵給 yfinance 或貼給使用者。
https://github.com/ranaroussi/yfinance ・ https://finance.yahoo.com/quote/USDJPY%3DX

拉價與產出外匯報告

我使用 yfinance 的 Ticker(symbol).history(period="5d") 拉近五天資料，計算日變動與趨勢，最後整成 Markdown 文字。
https://pypi.org/project/yfinance/ ・ https://github.com/ranaroussi/yfinance

def fetch_fx_quote_yf(symbol: str):
try:
tk = yf.Ticker(symbol)
df = tk.history(period="5d", interval="1d")
if df is None or df.empty:
return None, None, None, None
last_row = df.iloc[-1]
prev_row = df.iloc[-2] if len(df) >= 2 else None
last_price = float(last_row["Close"])
change_pct = None if prev_row is None else (last_price / float(prev_row["Close"]) - 1.0) * 100.0
ts = last_row.name
ts_iso = ts.tz_convert("Asia/Taipei").strftime("%Y-%m-%d %H:%M %Z") if hasattr(ts, "tz_convert") else str(ts)
return last_price, change_pct, ts_iso, df
except Exception as e:
logger.error(f"fetch_fx_quote_yf error for {symbol}: {e}", exc_info=True)
return None, None, None, None

def render_fx_report(base, quote, link, last, chg, ts, df) -> str:
trend = ""
if df is not None and not df.empty:
diff = float(df["Close"].iloc[-1]) - float(df["Close"].iloc[0])
trend = "上升" if diff > 0 else ("下跌" if diff < 0 else "持平")
lines = []
lines.append(f"#### 外匯報告（查匯優先）\n- 幣別對：{base}/{quote}\n- 來源：Yahoo Finance\n- 連結：{link}")
if last is not None: lines.append(f"- 目前匯率：{last:.6f}（{base}/{quote}）")
if chg is not None: lines.append(f"- 日變動：{chg:+.2f}%")
if ts: lines.append(f"- 資料時間：{ts}")
if trend: lines.append(f"- 近 5 日趨勢：{trend}")
lines.append("\n外匯連結（Yahoo）")
return "\n".join(lines)

https://pandas.pydata.org/docs/ ・ https://finance.yahoo.com/

讓 LINE 訊息「每則回覆都附 Quick Reply + 語音」

在我們的 reply_text_with_tts_and_extras() 統一出口中：先組 TextMessage 並附上 Quick Reply；接著若 CLOUDINARY_URL 存在且 TTS_SEND_ALWAYS=true，就把文字轉語音、上傳 Cloudinary，最後加一則 AudioMessage 一起回。
https://developers.line.biz/en/reference/messaging-api/#send-reply-message ・ https://cloudinary.com/documentation/image_and_video_upload_api_reference

[NEW] 單一出口：每次回覆都會自動附上 QuickReply 與（可選）TTS

async def reply_text_with_tts_and_extras(reply_token: str, text: str, extras: Optional[List]=None):
if not text:
text = "（無內容）"
messages = [attach_quick_reply(TextMessage(text=text))]
if extras:
for m in extras:
attach_quick_reply(m)
messages.extend(extras)

if os.getenv("TTS_SEND_ALWAYS", "true").lower() == "true" and CLOUDINARY_URL:
    try:
        audio_bytes = await text_to_speech_async(text)  # 內含 openai→gTTS 的自動 fallback
        if audio_bytes:
            def _upload():
                return cloudinary.uploader.upload(
                    io.BytesIO(audio_bytes),
                    resource_type="video",
                    folder="line-bot-tts",
                    format="mp3"
                )
            res = await run_in_threadpool(_upload)
            url = res.get("secure_url")
            if url:
                est = max(3000, min(30000, len(text) * 60))  # 依字數粗估語音長度
                messages.append(attach_quick_reply(AudioMessage(original_content_url=url, duration=est)))
    except Exception as e:
        logger.warning(f"TTS 附加失敗：{e}")  # 不阻擋主流程

# 注意：LINE v3 Python SDK 的 AsyncMessagingApi 在此回傳同步 Response，**不要 await**
try:
    line_bot_api.reply_message(ReplyMessageRequest(reply_token=reply_token, messages=messages))
except Exception as e:
    logger.error(f"LINE reply_message 失敗：{e}")

https://developers.line.biz/en/docs/messaging-api/using-quick-reply/ ・ https://github.com/line/line-bot-sdk-python/tree/master/v3

⸻

Page 2/2

常見錯誤與排查清單（今天踩到的坑）
1. TypeError: object ReplyMessageResponse can't be used in 'await' expression
代表你把 LINE v3 SDK 的回覆方法「當成 coroutine」去 await 了；實際上它是同步回傳，請改成不要 await。
https://github.com/line/line-bot-sdk-python/tree/master/v3
2. Invalid reply token
• 你可能在同一個 reply token 重複回覆或是回覆超過時間；或先前 await 例外被吃掉、重試時 token 已失效。
• 解法：確保每次事件只呼叫一次 reply_message，若要補第二則可改用 push_message 或在同一次 reply 放多個 message。
https://developers.line.biz/en/docs/messaging-api/building-bot/#webhook-event-objects
3. OpenAI TTS 401（金鑰錯或無權限）
• 設 OPENAI_API_KEY 後再測；或讓 TTS_PROVIDER=auto，自動 fallback 到 gTTS。
https://platform.openai.com/account/api-keys ・ https://platform.openai.com/docs/guides/text-to-speech
4. Cloudinary 上傳失敗
• 檢查 CLOUDINARY_URL；我建議用 resource_type="video" 與 format="mp3"，再回傳 secure_url。
https://cloudinary.com/documentation/image_upload_api_reference#upload
5. yfinance 外匯抓不到資料
• 確認代碼是否為 =X 後綴，像 USDJPY=X、JPYTWD=X；並留意免費源的延遲與缺漏。
https://github.com/ranaroussi/yfinance ・ https://finance.yahoo.com/

完整事件路由的核心邏輯（節錄，含人設切換、翻譯與外匯優先）

以下是 handle_text_message() 中最關鍵的決策流，確保「先翻譯模式→主選單→人設→金價/彩券→外匯優先→股票→一般聊天」的順序：
https://fastapi.tiangolo.com/ ・ https://developers.line.biz/en/reference/messaging-api/#send-reply-message

決策順序（節錄，含註解）

if m := TRANSLATE_CMD.match(msg): # 翻譯模式 on/off
...
elif msg.startswith("翻譯->"):
...
elif im := INLINE_TRANSLATE.match(msg): # 行內翻譯
...
elif _tstate_get(chat_id): # 翻譯模式持續中
...
elif low in ("menu", "選單", "主選單"): # 主選單
line_bot_api.reply_message(ReplyMessageRequest(reply_token=reply_tok, messages=[build_main_menu()]))
return
elif msg in PERSONA_ALIAS.keys(): # 人設切換（甜/鹹/萌/酷/random）
key = set_user_persona(chat_id, msg)
p = PERSONAS[key]
await reply_text_with_tts_and_extras(reply_tok, f"已切換為「{p['title']}」模式～{p['emoji']}")
return
elif msg in ("金價","黃金"): # 金價
...
elif msg in ("大樂透","威力彩","539"): # 彩票
...
elif _is_fx_query(msg): # 外匯優先
base, quote, link = parse_fx_pair(msg)
symbol = f"{base}{quote}=X"
last, chg, ts, df = fetch_fx_quote_yf(symbol)
report = render_fx_report(base, quote, link, last, chg, ts, df)
await reply_text_with_tts_and_extras(reply_tok, report)
return
elif _is_stock_query(msg): # 股票/大盤
ticker, name_hint, link = _normalize_ticker_and_name(msg)
content_block, _ = await run_in_threadpool(build_stock_prompt_block, ticker, name_hint)
report = await run_in_threadpool(render_stock_report, ticker, link, content_block)
await reply_text_with_tts_and_extras(reply_tok, report)
return
else: # 一般聊天（套用人設）
...

https://developers.line.biz/en/docs/messaging-api/using-quick-reply/ ・ https://docs.python.org/3/library/functions.html#await

驗收流程（你可以照單逐一打勾）
• 在群組輸入 JPY → 回「外匯報告（JPY/TWD）」＋ Yahoo 連結，並附語音；Quick Reply 第一顆是「主選單」。
• 輸入 USD/JPY → 回「外匯報告（USD/JPY）」正常。
• 輸入 2330 或 NVDA → 走股票報告（StockGPT），不會誤判成外匯。
• 輸入 甜 / 鹹 / 萌 / 酷 / random → 人設切換提示，後續聊天口吻改變。
• 長文回覆（> 200 字）仍會附語音；若關閉或金鑰錯誤，不會阻斷文字回覆（log 中可見 warning）。
https://developers.line.biz/en/reference/messaging-api/#send-reply-message ・ https://render.com/docs/deploys

今天的結論

Day 25 我們把「意圖判斷精準（查匯優先）」與「互動體感完整（每則都有語音與主選單）」一次打包完成，也同步補了最容易踩的 SDK 同步/非同步誤用問題。明天可以開始做風險控管與監控（token 錯誤率、回覆延遲、Cloudinary 失敗率）以及更彈性的金融查詢語法（自然語言解析到 BASE/QUOTE）。
https://developers.line.biz/en/docs/messaging-api/overview/ ・ https://fastapi.tiangolo.com/ ・ https://github.com/ranaroussi/yfinance ・ https://cloudinary.com/documentation ・ https://platform.openai.com/docs/guides/text-to-speech ・ https://pypi.org/project/gTTS/